package day03.acc

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Author wsl
 * @Description 序列化方法和属性
 * 初始化在driver端,运行程序是在executor端 跨进程通信
 */
object Serializable01 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("s1 ").setMaster("local[*]")
    val sc = new SparkContext(conf)

    val rdd: RDD[String] = sc.makeRDD(Array("hello world", "hello spark", "hive", "atguigu"))

    //1。在executor端用到了drive端对象的一个函数，这个类也需要序列化。
    val search = new Search("hello")

    search.getMatch1(rdd).collect().foreach(println)

    // 2.在executor端用到了drive端对象的的属性，这个类也需要序列化.
    search.getMatch2(rdd).collect().foreach(println)

    sc.stop()
  }

}

class Search(query: String) extends Serializable {

  def isMatch(s: String): Boolean = {
    s.contains(this.query)
  }

  //this表示Search这个类的对象，程序在运行过程中需要将Search对象序列化以后传递到Executor端
  //函数序列化:类继承Serializable
  def getMatch1(rdd: RDD[String]): RDD[String] = {

    rdd.filter(this.isMatch)
  }


  // 属性序列化：① 类继承Serializable
  def getMatch2(rdd: RDD[String]): RDD[String] = {

    rdd.filter(x => x.contains(this.query))

    //②将类变量query赋值给局部变量
    val q = this.query
    rdd.filter(x => x.contains(q))

    //③ 讲class变成样例类，样例类默认是序列化的
  }
}